package com.ry.kafka.controller;

import com.alibaba.fastjson.JSONObject;
import com.ry.bean.People;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

//@Component
public class KafkaConsumerController {

    @KafkaListener(topics = {"topic1"})
    public void onMessage(ConsumerRecord record){
        System.out.println("主题："+record.topic() +" 分区："+record.partition()+" 偏移量："+record.offset()+" 消息内容："+record.value());
    }

    @KafkaListener(topics={"topic2"})
    public void onMessageObject(ConsumerRecord record){
        People p = JSONObject.parseObject(record.value().toString(), People.class);
//        System.out.println("主题："+record.topic() +" 分区："+record.partition()+" 偏移量："+record.offset()+" 消息内容："+record.value());
        System.out.println("主题："+record.topic() +" 分区："+record.partition()+" 偏移量："+record.offset());
        System.out.println(" 消息内容："+p.getId()+" "+p.getName()+" "+p.getAge()+" "+p.getCustomerId());
    }

}
